package com.poetic.network.flow;

import com.poetic.network.flow.aggregate.CountAgg;
import com.poetic.network.flow.domain.ItemViewCount;
import com.poetic.network.flow.domain.UserBehavior;
import com.poetic.network.flow.enums.BehaviorEnum;
import com.poetic.network.flow.function.WindowResultFunction;
import com.poetic.network.flow.process.TopNHotItems;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * <pre>
 *
 * 我们将实现一个“实时热门商品”的需求，可以将“实时热门商品”翻译成程
 * 序员更好理解的需求：每隔 5 分钟输出最近一小时内点击量最多的前 N 个商品。将
 * 这个需求进行分解我们大概要做这么几件事情：
 * • 抽取出业务时间戳，告诉 Flink 框架基于业务时间做窗口
 * • 过滤出点击行为数据
 * • 按一小时的窗口大小，每 5 分钟统计一次，做滑动窗口聚合（Sliding Window）
 * • 按每个窗口聚合，输出每个窗口中点击量前 N 名的商品
 *
 * Created by lianghuikun on 2020-09-14.
 * </pre>
 *
 * @author lianghuikun
 */
public class HotItemsTask {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设定 Time 类型为 EventTime
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // 为了打印到控制台的结果不乱序，我们配置全局的并发为 1，这里改变并发对结果正确性没有影响
        env.setParallelism(1);
        String path = "/Users/lianghuikun/indigo/poetic/flink/user-behavior-anylysis/data/UserBehavior.csv";
        SingleOutputStreamOperator<UserBehavior> dataStream = env.readTextFile(path)
                .map(new MapFunction<String, UserBehavior>() {
                    @Override
                    public UserBehavior map(String value) throws Exception {
                        String[] data = value.split("\\,");
                        return UserBehavior.builder()
                                .userId(Long.valueOf(data[0]))
                                .itemId(Long.valueOf(data[1]))
                                .categoryId(Integer.valueOf(data[2]))
                                .behavior(data[3])
                                .timestamp(Long.valueOf(data[4]))
                                .build();
                    }
                });
        SingleOutputStreamOperator<ItemViewCount> processStream = dataStream
                // 过滤出点击事件
                .filter(new FilterFunction<UserBehavior>() {
                    @Override
                    public boolean filter(UserBehavior value) throws Exception {
                        return value.getBehavior().equals(BehaviorEnum.pv.name());
                    }
                })
                // 指定时间戳和 watermark
                .assignTimestampsAndWatermarks(WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                        .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                            @Override
                            public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                return element.getTimestamp() * 1000;
                            }
                        }))
                // 每隔 5 分钟统计一次最近一小时每个商品的点击量
                .keyBy(new KeySelector<UserBehavior, Long>() {
                    @Override
                    public Long getKey(UserBehavior value) throws Exception {
                        return value.getItemId();
                    }
                })
                .timeWindow(Time.hours(1), Time.minutes(5))
// 2.2.4
                /**
                 * aggregate 提 前 聚 合 掉 数 据 ， 减 少 state 的 存 储 压 力 。
                 * 较之 .apply(WindowFunction wf) 会将窗口中的数据都存储下来，最后一起计算要高效地多。
                 */
                .aggregate(new CountAgg(), new WindowResultFunction());
//                .print();

        /**
         * 为了统计每个窗口下最热门的商品，我们需要再次按窗口进行分组，这里根据
         * ItemViewCount 中的 windowEnd 进行 keyBy()操作。然后使用 ProcessFunction 实现
         * 一个自定义的 TopN 函数 TopNHotItems 来计算点击量排名前 3 名的商品，并将排名
         * 结果格式化成字符串，便于后续输出。
         */
        // 分组
        processStream.keyBy(new KeySelector<ItemViewCount, Long>() {
            @Override
            public Long getKey(ItemViewCount value) throws Exception {
                return value.getWindowEnd();
            }
        })
                // 排序
                .process(new TopNHotItems(3))
                .print();

//        TimeUnit.SECONDS.sleep(1000);
        env.execute("Hot Items Job");
    }
}
